package cn.doitedu.dwh

import org.apache.spark.sql.{DataFrame, SparkSession}

import scala.collection.immutable
import scala.collection.mutable.ListBuffer

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-06-14
 * @desc 事件归因分析
 *
 *    -- 需求举例：
 *       目标业务事件：  e8(p2=v3)
 *       待归因事件： e2(p1=v1)   e4(p2=v2)  e6
 *       归因策略： 首次触点归因
 *
 *   -- 源表： dwd.app_event_detail
 *   -- 模拟测试数据
 *       1,e1,p1=v2,1
 *       1,e3,p1=v2,2
 *       1,e4,p2=v2,3
 *       1,e2,p1=v1,4
 *       1,e6,p1=v2,5
 *       1,e8,p2=v3,6
 *       1,e2,p1=v1,7
 *       1,e1,p1=v2,8
 *       2,e3,p1=v2,11
 *       2,e4,p2=v2,12
 *       2,e5,p1=v2,13
 *       2,e2,p1=v1,14
 *       2,e6,p1=v2,15
 *       2,e8,p2=v3,16
 *       2,e2,p1=v1,17
 *       2,e6,p1=v2,18
 *       2,e8,p2=v3,19
 *
 *       drop table if exists dwd.app_event_detail_test_attr;
 *       create table dwd.app_event_detail_test_attr(
 *         guid  bigint,
 *         eventid string,
 *         properties Map<string,string>,
 *         `timestamp`  bigint
 *       )
 *       row format delimited fields terminated by ','
 *       collection items terminated by ':'
 *       map keys terminated by '='
 *       ;
 *
 *       load data local inpath '/root/attr.txt' into table  dwd.app_event_detail_test_attr;
 *
 *       -- 计算思路
 *       -- 首先，过滤出有用的事件，然后按用户分组，将每个用户的事件收集成数组
 *       u01,[e4,e2,e6,e8,e2]
 *       u02,[e4,e2,e6,e8,e2,e6,e8,e7]
 *
 *       -- 然后按目标事件E8，将行为序列分割成段
 *       u01,[[e4,e2,e6]]
 *       u02,[[e4,e2,e6],[e2,e6]]
 *
 *       -- 然后，将一个用户的多段行为序列，打散explode
 *       u01,[e4,e2,e6]
 *       u02,[e4,e2,e6]
 *       u02,[e2,e6]
 *
 *       -- 然后，根据需求中指定的策略
 *       -- （首次触点，末次触点，位置归因，线性归因，时间衰减归因），
 *       -- 开发UDF对数组进行计算即可
 */
object EventAttrbute {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName("事件归因分析计算任务")
      .master("local")
      .config("spark.sql.shuffle.partitions","2")
      .enableHiveSupport()
      .getOrCreate()


    // 行为序列分割函数
    val segment = (arr:Array[String]) =>
    {
      // 对事件按时间戳排序，并去掉时间戳
      val events = arr.sortBy(e => e.split("_")(0)).map(e => e.split("_")(1))

      // 过滤掉最后一个e8后的事件
      val i = events.lastIndexOf("e8")
      val newArray = events.slice(0, i + 1)

      var segements = new ListBuffer[Array[String]]()
      var segement = new ListBuffer[String]()

      for (elem <- newArray) {
        if(elem.equals("e8")){
          // 如果元素是e8，并将之前”段“存入”段列表“
          segements.+=(segement.toArray)
          // 新建一个“段“，
          segement = new ListBuffer[String]()
        }else{
          // 如果元素不是e8，则将该元素（事件）加入“段”
          segement.+=(elem)
        }
      }
      segements.toArray
    }
    // 注册分割寒暑
    spark.udf.register("seg",segment)

    // 调用首次触点归因分析
    val res1 = firstEventAttribute(spark)
    //res1.show(100,false)


    // 调用末次触点归因分析
    val res2 = lastEventAttribute(spark)
    //res2.show(100,false)

    // 线性归因分析  [e2,e4,e6]  => e2:0.33,  e4:0.33  ,e6:0.33

    // 位置归因分析 [e2,e6,e5,e4] => e2:0.4,   e6:0.1,e5:0.1,    e4:0.4

    // 时间衰减归因分析  [e2,e6,e5,e4] =>
    //       衰减权重： 0.72,0.81,0.9,1
    // 影响力的结果： 0.72 /(0.72+0.81+0.9+1)   e2:0.21,e6:0.24,e5:0.26,e6:0.29
    val res3 = timeDecayEventAttribute(spark)
    res3.show(100,false)


    spark.close()

  }

  /**
   * 公共方法，用于：
   * 将每个用户的行为序列，按照目标行为事件进行分段
   * @param spark
   * @return
   */
  def eventSplit(spark:SparkSession)={
    spark.sql(
      """
        |with tmp as (
        |select
        |  guid,
        |  seg(collect_list(concat_ws('_',timestamp,eventid))) segments
        |from dwd.app_event_detail_test_attr
        | where
        | -- dt='20201-06-06' and
        |(
        |     (eventid='e8' and properties['p2']='v3')
        |  or (eventid='e2' and properties['p1']='v1')
        |  or (eventid='e4' and properties['p2']='v2')
        |  or (eventid='e6')
        |)
        |group by guid
        |)
        |
        |select
        |  guid,
        |  segment
        |from tmp  lateral view explode(segments) o as segment
        |
        |""".stripMargin)
  }


  // 首次触点归因
  def firstEventAttribute(spark:SparkSession): DataFrame ={
    val segmentsDataframe = eventSplit(spark)
    val res = segmentsDataframe.selectExpr("guid","segment[0] as first_attr_event")
    res
  }


  // 末次触点归因
  def lastEventAttribute(spark:SparkSession): DataFrame ={
    val segmentsDataframe = eventSplit(spark)
    val res = segmentsDataframe.selectExpr("guid","segment[size(segment)-1] as last_attr_event")
    res
  }


  // 时间衰减归因
  def timeDecayEventAttribute(spark:SparkSession):DataFrame = {

    // 行为序列分割
    val segmentsDataframe = eventSplit(spark)

    // 时间衰减归因权重计算
    val timeDecayAttr = (arr:Array[String])=>{
      val reversed = arr.reverse
      val fenmu =  (for(i <- 0 until arr.size) yield Math.pow(0.9,i) ).sum

      // ["e2_0.23", "e6_0.36", "e4_0.51"]
      for(i <- 0 until arr.size) yield {
        // "e2_0.23"
        arr(i) + "_"+Math.pow(0.9,arr.size-1-i)/fenmu
      }
    }

    spark.udf.register("time_decay_attr",timeDecayAttr)

    segmentsDataframe.createTempView("tmp")

    val res = spark.sql(
      """
        |select
        |   guid,
        |   split(event_attr,'_')[0] as event,
        |   cast(split(event_attr,'_')[1] as double) as attr
        |from tmp  lateral view explode(time_decay_attr(segment)) o as event_attr
        |
        |""".stripMargin)

    res

  }


}
